package com.alison.datastream.chapter4_join;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * @Author alison
 * @Date 2024/4/5 17:30
 * @Version 1.0
 * @Description
 */
public class F2_IntervalJoinExample {

    // 数据流区间 join , intervalJoin
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //只能使用 event_time
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        DataStream<Tuple3<String, Long, Integer>> input1 = env.addSource(new MySource(1))
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<Tuple3<String, Long, Integer>>forBoundedOutOfOrderness(Duration.ofSeconds(1))
                        .withTimestampAssigner(((element, recordTimestamp) -> element.f1)));
        input1.print("input1");

        DataStream<Tuple3<String, Long, Integer>> input2 = env.addSource(new MySource(2))
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<Tuple3<String, Long, Integer>>forBoundedOutOfOrderness(Duration.ofSeconds(1))
                        .withTimestampAssigner(((element, recordTimestamp) -> element.f1)));
        input2.print("input2");

        SingleOutputStreamOperator<String> intervelJoinResult = input1.keyBy(i -> i.f0)
                .intervalJoin(input2.keyBy(i -> i.f0))
                .between(Time.seconds(-5), Time.seconds(10))
                .process(new ProcessJoinFunction<Tuple3<String, Long, Integer>, Tuple3<String, Long, Integer>, String>() {
                    @Override
                    public void processElement(Tuple3<String, Long, Integer> left, Tuple3<String, Long, Integer> right,
                                               Context ctx,
                                               Collector<String> out) throws Exception {
                        out.collect("input1: " + left.toString() + ", input2: " + right.toString());
                    }
                });
        intervelJoinResult.print("result");
        env.execute("interval join function");


    }


    // key, 时间戳, 值
    public static class MySource implements SourceFunction<Tuple3<String, Long, Integer>> {
        private volatile boolean isRunning = true;

        int count = 1;

        public MySource(int count) {
            this.count = count;
        }

        @Override
        public void run(SourceContext<Tuple3<String, Long, Integer>> ctx) throws Exception {

            while (isRunning) {
                Tuple3<String, Long, Integer> data = Tuple3.of(String.valueOf(count), System.currentTimeMillis(), count);
                ctx.collect(data);
                count++;
                TimeUnit.SECONDS.sleep(2);
            }
        }

        @Override
        public void cancel() {
            isRunning = false;
        }
    }
}
